Глава 2: Маршрутизация

Обзор паттерна «Маршрутизация»

Последовательная обработка через сцепление промптов (prompt chaining) — фундаментальная техника для детерминированных линейных рабочих процессов с языковыми моделями, однако её применимость ограничена там, где требуются адаптивные ответы. Реальные агентные системы часто должны выбирать между несколькими возможными действиями в зависимости от внешних обстоятельств: состояния среды, пользовательского ввода или результата предыдущей операции. Эта способность к динамическому принятию решений, управляющая направлением потока к специализированным функциям, инструментам или подпроцессам, достигается с помощью механизма, известного как маршрутизация.

Маршрутизация вводит в операционную структуру агента условную логику, переводя его поведение от фиксированного пути исполнения к модели, где агент динамически оценивает заданные критерии и выбирает один из нескольких возможных последующих шагов. Это обеспечивает более гибкое и контекстно‑зависимое поведение системы.

Например, агент для обработки пользовательских обращений, оснащённый функцией маршрутизации, сначала может классифицировать входящий запрос для определения намерения пользователя. На основе этой классификации он направит запрос специализированному агенту для прямого ответа на вопрос, инструменту извлечения из базы данных для информации об аккаунте или включит процедуру эскалации для сложных случаев — вместо использования единственного заранее заданного маршрута. Таким образом, более развитый агент с маршрутизацией может:

1. Проанализировать запрос пользователя.
2. Маршрутизировать запрос по его намерению:
   • Если намерение — «проверить статус заказа», направить в подагента или цепочку инструментов, взаимодействующих с базой заказов.
   • Если намерение — «информация о продукте», направить в подагента или цепочку, ищущую по каталогу товаров.
   • Если намерение — «техническая поддержка», направить в иную цепочку, использующую руководства по устранению неполадок или эскалацию к человеку.
   • Если намерение неясно, направить в подагента уточнения или соответствующую цепочку промптов.

Ключевой компонент паттерна «Маршрутизация» — механизм, выполняющий оценку и управляющий потоком. Его можно реализовать несколькими способами:

• Маршрутизация на базе LLM. Сама языковая модель может быть приглашена проанализировать вход и выдать конкретный идентификатор/инструкцию, указывающую следующий шаг или адресата. Например: «Проанализируй запрос пользователя и выдай только категорию: ‘Order Status’, ‘Product Info’, ‘Technical Support’ или ‘Other’». Система считывает этот ответ и направляет выполнение соответствующим образом.
• Маршрутизация на основе эмбеддингов. Входной запрос переводится в векторное представление (см. RAG, глава 14), затем сравнивается с векторами, соответствующими различным маршрутам/возможностям. Запрос направляется по маршруту с наибольшей семантической близостью. Это полезно для семантической маршрутизации, где решение основывается на смысле, а не ключевых словах.
• Правил‑ориентированная маршрутизация. Использование заранее заданных правил/логики (if‑else, switch), основанных на ключевых словах, паттернах или структурированных данных, извлечённых из входа. Такой подход быстрее и детерминированнее, чем LLM‑маршрутизация, но хуже обобщает новые/тонкие случаи.
• Маршрутизация на основе модели машинного обучения. Применяется дискриминативная модель (классификатор), специально обученная на небольшом размеченном корпусе для задачи маршрутизации. Хотя концептуально похожа на методы с эмбеддингами, её отличает супервайз‑дообучение, записывающее логику маршрутизации в веса модели. В отличие от LLM‑подхода, решение принимает не генеративная модель на инференсе, а заранее дообученная задача‑специфическая модель. LLM могут использоваться на этапе препроцессинга для синтетического расширения датасета, но не участвуют в принятии решения в реальном времени.

Механизмы маршрутизации могут применяться на разных этапах операционного цикла агента: в начале — для классификации главной задачи; на промежуточных шагах — для выбора последующего действия; внутри подпрограмм — для выбора наиболее подходящего инструмента из набора.

Каркасы вроде LangChain, LangGraph и Google Agent Developer Kit (ADK) предоставляют явные конструкции для описания и управления подобной условной логикой. Благодаря архитектуре графа состояний LangGraph особенно удобен для сложной маршрутизации, где решения зависят от накопленного состояния системы. Аналогично, ADK предоставляет базовые компоненты для структурирования возможностей и моделей взаимодействия агента, на которых строится логика маршрутизации. В рамках этих сред разработчики определяют возможные операционные пути и функции/модельные оценки, задающие переходы между узлами вычислительного графа.

Реализация маршрутизации позволяет выйти за рамки детерминированной последовательной обработки. Она поддерживает построение более адаптивных потоков исполнения, реагирующих динамично и уместно на широкий спектр входов и изменений состояния.

Практические применения и сценарии

Паттерн «Маршрутизация» — критически важный механизм управления в адаптивных агентных системах: он позволяет динамически менять путь исполнения в ответ на вариативные входы и внутренние состояния. Его полезность охватывает множество доменов, обеспечивая необходимый слой условной логики.

В HCI‑сценариях (виртуальные ассистенты, ИИ‑тьюторы) маршрутизация используется для интерпретации намерений. Первичный анализ естественно‑языкового запроса определяет последующее действие: вызов конкретного инструмента поиска, эскалация к оператору или выбор следующего модуля учебной программы с учётом прогресса пользователя. Это выводит систему за пределы линейных диалогов и делает ответы контекстными.

В автоматизированных конвейерах обработки данных и документов маршрутизация выступает как функция классификации и распределения. Входящие данные (письма, тикеты, API‑полезные нагрузки) анализируются по содержимому, метаданным или формату; далее элементы направляются в соответствующие рабочие процессы: обработка лидов, специфические трансформации для JSON/CSV, путь срочной эскалации и т. п.

В комплексных системах со множеством инструментов/агентов маршрутизация действует как диспетчер верхнего уровня. Исследовательская система со специализированными агентами для поиска, суммаризации и анализа использует роутер, чтобы назначать задачи наиболее подходящему агенту под текущую цель. Аналогично, ИИ‑ассистент программирования сначала определяет язык и намерение пользователя — отладка, объяснение, перевод — и только затем передаёт фрагмент кода в соответствующий специализированный инструмент.

В конечном счёте маршрутизация даёт возможность логического арбитража, необходимого для создания функционально разнообразных и контекстно‑осведомлённых систем. Агент перестаёт быть статическим исполнителем предопределённых последовательностей и становится динамической системой, принимающей решения о наиболее эффективном способе выполнения задачи при меняющихся условиях.

Практический пример кода (LangChain)

Реализация маршрутизации в коде предполагает определение возможных путей и логики выбора между ними. Фреймворки LangChain и LangGraph предоставляют для этого специализированные компоненты и структуры. Граф состояний LangGraph особенно нагляден для визуализации и реализации маршрутизации.

Ниже показан простой «агентоподобный» пример на LangChain и Google Generative AI. Он создаёт «координатора», который маршрутизирует пользовательские запросы к симулированным «подагентам» по намерению (бронирование, информация, неясно). Система использует языковую модель для классификации запроса и делегирует его соответствующему обработчику, моделируя базовый паттерн делегирования, типичный для мультиагентных архитектур.

Сначала установите необходимые библиотеки:

pip install langchain langgraph google-cloud-aiplatform langchain-google-genai google-adk deprecated pydantic

Также потребуется настроить окружение с API‑ключом выбранной модели (например, OpenAI, Google Gemini, Anthropic).

# Copyright (c) 2025 Marco Fago
# https://www.linkedin.com/in/marco-fago/
#
# This code is licensed under the MIT License.
# See the LICENSE file in the repository for the full license text.

from langchain_google_genai import ChatGoogleGenerativeAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import RunnablePassthrough, RunnableBranch

# --- Configuration ---
# Ensure your API key environment variable is set (e.g., GOOGLE_API_KEY)
try:
    llm = ChatGoogleGenerativeAI(model="gemini-2.5-flash", temperature=0)
    print(f"Language model initialized: {llm.model}")
except Exception as e:
    print(f"Error initializing language model: {e}")
    llm = None

# --- Define Simulated Sub-Agent Handlers (equivalent to ADK sub_agents) ---
def booking_handler(request: str) -> str:
    """Simulates the Booking Agent handling a request."""
    print("\n--- DELEGATING TO BOOKING HANDLER ---")
    return f"Booking Handler processed request: '{request}'. Result: Simulated booking action."

def info_handler(request: str) -> str:
    """Simulates the Info Agent handling a request."""
    print("\n--- DELEGATING TO INFO HANDLER ---")
    return f"Info Handler processed request: '{request}'. Result: Simulated information retrieval."

def unclear_handler(request: str) -> str:
    """Handles requests that couldn't be delegated."""
    print("\n--- HANDLING UNCLEAR REQUEST ---")
    return f"Coordinator could not delegate request: '{request}'. Please clarify."

# --- Define Coordinator Router Chain (equivalent to ADK coordinator's instruction) ---
# This chain decides which handler to delegate to.
coordinator_router_prompt = ChatPromptTemplate.from_messages([
    ("system", """Analyze the user's request and determine which specialist handler should process it.
     - If the request is related to booking flights or hotels,
        output 'booker'.
     - For all other general information questions, output 'info'.
     - If the request is unclear or doesn't fit either category,
        output 'unclear'.
     ONLY output one word: 'booker', 'info', or 'unclear'."""),
    ("user", "{request}")
])

if llm:
    coordinator_router_chain = coordinator_router_prompt | llm | StrOutputParser()

# --- Define the Delegation Logic (equivalent to ADK's Auto-Flow based on sub_agents) ---
# Use RunnableBranch to route based on the router chain's output.

# Define the branches for the RunnableBranch
branches = {
    "booker": RunnablePassthrough.assign(output=lambda x: booking_handler(x['request']['request'])),
    "info": RunnablePassthrough.assign(output=lambda x: info_handler(x['request']['request'])),
    "unclear": RunnablePassthrough.assign(output=lambda x: unclear_handler(x['request']['request'])),
}

# Create the RunnableBranch. It takes the output of the router chain
# and routes the original input ('request') to the corresponding handler.
delegation_branch = RunnableBranch(
    (lambda x: x['decision'].strip() == 'booker', branches["booker"]), # Added .strip()
    (lambda x: x['decision'].strip() == 'info', branches["info"]),     # Added .strip()
    branches["unclear"] # Default branch for 'unclear' or any other output
)

# Combine the router chain and the delegation branch into a single runnable
# The router chain's output ('decision') is passed along with the original input ('request')
# to the delegation_branch.
coordinator_agent = {
    "decision": coordinator_router_chain,
    "request": RunnablePassthrough()
} | delegation_branch | (lambda x: x['output']) # Extract the final output

# --- Example Usage ---
def main():
    if not llm:
        print("\nSkipping execution due to LLM initialization failure.")
        return

    print("--- Running with a booking request ---")
    request_a = "Book me a flight to London."
    result_a = coordinator_agent.invoke({"request": request_a})
    print(f"Final Result A: {result_a}")

    print("\n--- Running with an info request ---")
    request_b = "What is the capital of Italy?"
    result_b = coordinator_agent.invoke({"request": request_b})
    print(f"Final Result B: {result_b}")

    print("\n--- Running with an unclear request ---")
    request_c = "Tell me about quantum physics."
    result_c = coordinator_agent.invoke({"request": request_c})
    print(f"Final Result C: {result_c}")

if __name__ == "__main__":
    main()

Как отмечено выше, этот код на Python конструирует простой «агентоподобный» контур с использованием LangChain и модели Google Generative AI (gemini‑2.5‑flash). Определены три симулированных обработчика: booking_handler, info_handler и unclear_handler, каждый предназначен для определённого рода запросов.

Ключевой компонент — coordinator_router_chain, использующий ChatPromptTemplate для инструкции модели: классифицировать входящий запрос в одну из категорий — ‘booker’, ‘info’ или ‘unclear’. Выход этой цепочки поступает в RunnableBranch, который делегирует исходный запрос соответствующей функции‑обработчику. Объединённый coordinator_agent сначала получает решение маршрутизатора, затем передаёт запрос выбранному обработчику и извлекает итоговый ответ.

Функция main демонстрирует работу на трёх примерах, показывая, как разные запросы маршрутизируются и обрабатываются подагентами. Добавлена обработка ошибок инициализации языковой модели.

Практический пример кода (Google ADK)

Agent Development Kit (ADK) — фреймворк для инженерии агентных систем, предоставляющий структурированную среду для определения возможностей и поведения агента. В отличие от архитектур явных вычислительных графов, в ADK маршрутизация обычно реализуется через набор «инструментов», представляющих функции агента. Выбор подходящего инструмента по пользовательскому запросу выполняется внутренней логикой фреймворка, использующей модель для сопоставления намерения с обработчиком.

Ниже приведён пример приложения на Google ADK. Создаётся агент‑«Координатор», маршрутизирующий запросы пользователям к специализированным подагентам («Booker» для бронирований и «Info» для общих сведений) на основе заданных инструкций. Подагенты используют инструменты для симуляции обработки запросов, демонстрируя базовый паттерн делегирования в агентной системе.

# Copyright (c) 2025 Marco Fago
#
# This code is licensed under the MIT License.
# See the LICENSE file in the repository for the full license text.

import uuid
from typing import Dict, Any, Optional
from google.adk.agents import Agent
from google.adk.runners import InMemoryRunner
from google.adk.tools import FunctionTool
from google.genai import types
from google.adk.events import Event

# --- Define Tool Functions ---
# These functions simulate the actions of the specialist agents.
def booking_handler(request: str) -> str:
    """
    Handles booking requests for flights and hotels.
    Args:
        request: The user's request for a booking.
    Returns:
        A confirmation message that the booking was handled.
    """
    print("-------------------------- Booking Handler Called ----------------------------")
    return f"Booking action for '{request}' has been simulated."

def info_handler(request: str) -> str:
    """
    Handles general information requests.
    Args:
        request: The user's question.
    Returns:
        A message indicating the information request was handled.
    """
    print("-------------------------- Info Handler Called ----------------------------")
    return f"Information request for '{request}'. Result: Simulated information retrieval."

def unclear_handler(request: str) -> str:
    """Handles requests that couldn't be delegated."""
    return f"Coordinator could not delegate request: '{request}'. Please clarify."

# --- Create Tools from Functions ---
booking_tool = FunctionTool(booking_handler)
info_tool = FunctionTool(info_handler)

# Define specialized sub-agents equipped with their respective tools
booking_agent = Agent(
    name="Booker",
    model="gemini-2.0-flash",
    description="A specialized agent that handles all flight "
                "and hotel booking requests by calling the booking tool.",
    tools=[booking_tool]
)

info_agent = Agent(
    name="Info",
    model="gemini-2.0-flash",
    description="A specialized agent that provides general information "
                "and answers user questions by calling the info tool.",
    tools=[info_tool]
)

# Define the parent agent with explicit delegation instructions
coordinator = Agent(
    name="Coordinator",
    model="gemini-2.0-flash",
    instruction=(
        "You are the main coordinator. Your only task is to analyze "
        "incoming user requests "
        "and delegate them to the appropriate specialist agent. "
        "Do not try to answer the user directly.\n"
        "- For any requests related to booking flights or hotels, "
        "delegate to the 'Booker' agent.\n"
        "- For all other general information questions, delegate to the 'Info' agent."
    ),
    description="A coordinator that routes user requests to the "
                "correct specialist agent.",
    # The presence of sub_agents enables LLM-driven delegation (Auto-Flow) by default.
    sub_agents=[booking_agent, info_agent]
)

# --- Execution Logic ---
async def run_coordinator(runner: InMemoryRunner, request: str):
    """Runs the coordinator agent with a given request and delegates."""
    print(f"\n--- Running Coordinator with request: '{request}' ---")
    final_result = ""
    try:
        user_id = "user_123"
        session_id = str(uuid.uuid4())
        await runner.session_service.create_session(
            app_name=runner.app_name, user_id=user_id, session_id=session_id
        )
        for event in runner.run(
            user_id=user_id,
            session_id=session_id,
            new_message=types.Content(
                role='user',
                parts=[types.Part(text=request)]
            ),
        ):
            if event.is_final_response() and event.content:
                # Try to get text directly from event.content
                # to avoid iterating parts
                if hasattr(event.content, 'text') and event.content.text:
                    final_result = event.content.text
                elif event.content.parts:
                    # Fallback: Iterate through parts and extract text (might trigger warning)
                    text_parts = [part.text for part in event.content.parts if part.text]
                    final_result = "".join(text_parts)
                # Assuming the loop should break after the final response
                break
        print(f"Coordinator Final Response: {final_result}")
        return final_result
    except Exception as e:
        print(f"An error occurred while processing your request: {e}")
        return f"An error occurred while processing your request: {e}"

async def main():
    """Main function to run the ADK example."""
    print("--- Google ADK Routing Example (ADK Auto-Flow Style) ---")
    print("Note: This requires Google ADK installed and authenticated.")
    runner = InMemoryRunner(coordinator)

    # Example Usage
    result_a = await run_coordinator(runner, "Book me a hotel in Paris.")
    print(f"Final Output A: {result_a}")

    result_b = await run_coordinator(runner, "What is the highest mountain in the world?")
    print(f"Final Output B: {result_b}")

    result_c = await run_coordinator(runner, "Tell me a random fact.") # Should go to Info
    print(f"Final Output C: {result_c}")

    result_d = await run_coordinator(runner, "Find flights to Tokyo next month.") # Should go to Booker
    print(f"Final Output D: {result_d}")

if __name__ == "__main__":
    import nest_asyncio
    nest_asyncio.apply()
    await main()

Сценарий состоит из главного агента‑Координатора и двух специализированных подагентов: Booker и Info. Каждый подагент оснащён FunctionTool, оборачивающим Python‑функцию, симулирующую действие. booking_handler имитирует обработку бронирований, info_handler — извлечение общих сведений. unclear_handler служит запасным вариантом для случаев, когда делегирование невозможно (в текущей логике он явно не используется при сбое делегирования).

Главная роль Координатора — анализировать входящие сообщения и делегировать их либо Booker, либо Info. Делегирование автоматически реализовано механизмом Auto‑Flow в ADK, поскольку у Координатора определены sub_agents. Функция run_coordinator создаёт InMemoryRunner, формирует user_id/session_id и запускает обработку запроса через координатора. Метод runner.run выдаёт события, из которых код извлекает финальный текст ответа.

Функция main демонстрирует работу на нескольких примерах, показывая делегирование запросов о бронированиях агенту Booker, а информационных запросов — агенту Info.

Коротко

Что: Агентные системы сталкиваются с широким спектром входов и ситуаций, которые не укладываются в один линейный процесс. Простая последовательность не принимает решений по контексту. Без механизма выбора корректного инструмента или подпроцесса система остаётся жёсткой и неадаптивной, что затрудняет построение зрелых приложений для реальных пользовательских запросов.

Почему: Паттерн «Маршрутизация» вводит в операционный каркас слой условной логики. Система сперва анализирует входящий запрос, чтобы определить его природу/намерение, затем динамически направляет управление к соответствующему специализированному инструменту, функции или подагенту. Решение может приниматься при помощи LLM‑промптов, заранее заданных правил или семантической близости эмбеддингов. В результате статичный предопределённый путь превращается в гибкий контекстно‑осведомлённый рабочий процесс, выбирающий лучший следующий шаг.

Практическое правило: Применяйте маршрутизацию, когда агент должен выбирать между несколькими различными рабочими процессами, инструментами или подагентами на основе пользовательского ввода или текущего состояния. Это особенно важно для систем, которые должны сортировать входящие запросы по типам задач, например чат‑боты поддержки, различающие вопросы по продажам, технической поддержке и управлению аккаунтом.

Визуальное резюме

Рис. 1. Паттерн «Маршрутизация»: использование LLM в роли маршрутизатора

Ключевые выводы

• Маршрутизация позволяет агентам принимать динамические решения о следующем шаге на основе условий.
• Она даёт возможность обрабатывать разнообразные входы и адаптировать поведение, выходя за рамки линейного исполнения.
• Логику маршрутизации можно реализовывать с помощью LLM, правил или семантического сходства эмбеддингов.
• Фреймворки LangGraph и Google ADK предоставляют структурированные способы определения и управления маршрутизацией, хотя и с разными архитектурными подходами.

Заключение

Паттерн «Маршрутизация» — ключевой шаг на пути к по‑настоящему динамичным и отзывчивым агентным системам. Реализуя маршрутизацию, мы выходим за пределы простых линейных потоков и наделяем агентов способностью разумно решать, как обрабатывать информацию, отвечать на запросы и использовать доступные инструменты или подагентов.

Мы увидели, как маршрутизация применима в различных доменах — от чат‑ботов поддержки до конвейеров обработки данных. Способность анализировать вход и условно направлять рабочий процесс — фундамент для агентов, справляющихся с естественной вариативностью реальных задач.

Примеры на LangChain и Google ADK демонстрируют два разных, но эффективных подхода. Граф‑ориентированный LangGraph обеспечивает наглядное явное определение состояний и переходов, что удобно для сложных многошаговых рабочих процессов с развитой маршрутизацией. ADK, напротив, фокусируется на определении дискретных возможностей (инструментов) и полагается на способность фреймворка направлять запросы к подходящим обработчикам — это проще для агентов с чётко очерченным набором действий.

Освоение паттерна «Маршрутизация» необходимо для создания агентов, способных разумно ориентироваться в различных сценариях и выдавать адаптированные ответы/действия в зависимости от контекста. Это ключевой компонент в построении универсальных и надёжных агентных приложений.

Ссылки

1. Документация LangGraph
2. Документация Google Agent Developer Kit

---

Навигация

Назад: Глава 1. Цепочки промптов
Вперед: Глава 3. Параллелизация